package com.lagou.myrpcconsumer.proxy;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.lagou.myrpcconsumer.client.RpcClient;
import com.lagou.rpc.common.RpcRequest;
import com.lagou.rpc.common.RpcResponse;
import com.lagou.rpc.pojo.Ts;
import com.lagou.rpc.utils.GetZkClient;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.TreeCache;
import org.apache.curator.framework.recipes.cache.TreeCacheEvent;
import org.apache.zookeeper.data.Stat;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.lang.reflect.Proxy;
import java.time.LocalDateTime;
import java.util.*;

/**
 * 客户端代理类-创建代理对象
 * 1.封装request请求对象
 * 2.创建RpcClient对象
 * 3.发送消息
 * 4.返回结果
 */
@Component
public class RpcClientProxy {

    private static final List<RpcClient> clients = new ArrayList<>();

    // 保存对应服务器的时间戳 key值和clientMap相同
    public Map<String, Ts> tsMap = new HashMap<>();


    private static final Map<String, RpcClient> clientMap = new HashMap<>();

    //// 初始化两个客户端
    //static {
    //    clients.add(new RpcClient("127.0.0.1", 8091));
    //    clients.add(new RpcClient("127.0.0.1", 8092));
    //}


    /**
     * 轮询获取RpcClient对象
     *
     * @return
     */
    public RpcClient getRpcClient() {
        RpcClient rpcClient;
        rpcClient = clients.get(new Random().nextInt(clients.size()));
        return rpcClient;
    }


    /**
     * 创建代理类--发送消息
     *
     * @param serviceClass
     * @return
     */
    public Object createProxy(Class serviceClass) {
        return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(),
                new Class[]{serviceClass}, (proxy, method, args) -> {
                    //1.封装request请求对象
                    RpcRequest rpcRequest = new RpcRequest();
                    rpcRequest.setRequestId(UUID.randomUUID().toString());
                    rpcRequest.setClassName(method.getDeclaringClass().getName());
                    rpcRequest.setMethodName(method.getName());
                    rpcRequest.setParameterTypes(method.getParameterTypes());
                    rpcRequest.setParameters(args);
                    //2.创建RpcClient对象
                    //RpcClient rpcClient = new RpcClient("127.0.0.1", 8899);
                    // 根据之前调用动态获取rpc客户端
                    RpcClient rpcClient = getRpcClient();
                    try {
                        //3.发送消息
                        Object responseMsg = rpcClient.send(JSON.toJSONString(rpcRequest));
                        RpcResponse rpcResponse = JSON.parseObject(responseMsg.toString(), RpcResponse.class);
                        if (rpcResponse.getError() != null) {
                            throw new RuntimeException(rpcResponse.getError());
                        }
                        //4.返回结果
                        Object result = rpcResponse.getResult();
                        //打印
                        System.out.println("当前服务：" + rpcClient.getIp() + ":" + rpcClient.getPort());
                        // 调用成功后 计算时间 记录返回时间时间,记录调用使用时间
                        calcTime(rpcClient.getIp() + ":" + rpcClient.getPort(), System.currentTimeMillis());

                        return JSON.parseObject(result.toString(), method.getReturnType());
                    } catch (Exception e) {
                        throw e;
                    } finally {
                        //关闭管道和客户段线程
                        //rpcClient.close();
                    }

                });
    }

    // 从zk 当中获取服务端信息,然后放入到clientMap 当中并监听zk节点的变化
    @PostConstruct
    public void initZkClient() throws Exception {
        // 获取zk的实例
        CuratorFramework zkClient = GetZkClient.getZkClient();
        // 监听zk节点的变化
        watchNode(zkClient);
    }

    public void calcTime(String key, Long time) {

        // 获取zk的实例
        CuratorFramework zkClient = GetZkClient.getZkClient();
        Ts ts = tsMap.get(key);
        long latsTime = ts.getLatsTime();
        long useTime = time - latsTime;
        ts.setUseTime(useTime);
        ts.setLatsTime(System.currentTimeMillis());
        tsMap.put(key,ts);
        try {
            zkClient.setData().forPath(GetZkClient.path+"/"+key,JSONObject.toJSONString(ts).getBytes());
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 监控zookeeper节点的变化
     */
    private void watchNode(CuratorFramework zkClient) throws Exception {
        TreeCache treeCache = new TreeCache(zkClient, GetZkClient.path);
        treeCache.getListenable().addListener((curatorFramework, treeCacheEvent) -> {
            ChildData data = treeCacheEvent.getData();
            if (data == null) {
                return;
            }
            String str = data.getPath().substring(data.getPath().lastIndexOf("/") + 1);
            //127.0.0.1:8898

            System.out.println("str ===" + str);

            TreeCacheEvent.Type type = treeCacheEvent.getType();
            // 添加节点
            if (type == TreeCacheEvent.Type.NODE_ADDED) {
                String value = new String(data.getData());
                Ts ts = JSONObject.parseObject(value, Ts.class);
                System.out.println("ts.getLatsTime()====" + ts.getLatsTime());
                String[] split = str.split(":");
                if (split.length != 2) {
                    return;
                }

                if (!clientMap.containsKey(str)) {
                    RpcClient rpcClient = new RpcClient(split[0], Integer.parseInt(split[1]));
                    // 已经建立好了连接
                    clientMap.put(str, rpcClient);
                    // 将拥有的客户端存储起来
                    clients.add(rpcClient);
                    // 将时间戳存储起来
                    tsMap.put(str, ts);
                }
            } else if (type == TreeCacheEvent.Type.NODE_REMOVED) { // 节点删除
                System.out.println("被删除的节点：" + str);
                clientMap.remove(str);
                // 将拥有的客户端移除起来
                clients.remove(str);
                String value = new String(data.getData());
                // 将时间戳移除
                tsMap.remove(str, value);
            }

            System.out.println("============================");
            System.out.println("现存的节点");
            Set<String> strings = clientMap.keySet();
            for (String string : strings) {
                System.out.println("现有服务器为>>>" + string);
            }
        });

        treeCache.start();
    }


    @Scheduled(cron = "0/5 * * * * ?")
    //定时调用，超过五秒响应，就清零
    //@Scheduled(fixedRate=5000)
    private void configureTasks() {
        System.out.println("开始执行定时任务");
        for (Map.Entry<String, Ts> stringTsEntry : tsMap.entrySet()) {
            Ts value = stringTsEntry.getValue();
            if (value.getUseTime() > 5000) {
                value.setUseTime(1);
            }
        }
    }

}
